package com.gitee.wsl.coroutines.event

import kotlinx.coroutines.*
import kotlinx.coroutines.test.StandardTestDispatcher
import kotlinx.coroutines.test.UnconfinedTestDispatcher
import kotlinx.coroutines.test.runTest
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertSame

@Suppress("NAME_CONTAINS_ILLEGAL_CHARS")
class EventsTest {

    data class TestCoroutineEventA(val id: Any) : CoroutineEvent
    data class TestCoroutineEventB(val id: Any) : CoroutineEvent

    @Test
    fun `test - simple event emission - single collector`() = runTest(Events()) {
        val receivedEvents = mutableListOf<TestCoroutineEventA>()
        collectEventsAsync<TestCoroutineEventA>(Dispatchers.Unconfined) { receivedEvents += it }

        TestCoroutineEventA(1).emit()
        coroutineContext.eventsOrThrow.emit(TestCoroutineEventA(2))

        coroutineContext.cancelChildren()


        assertEquals(listOf(TestCoroutineEventA(1), TestCoroutineEventA(2)), receivedEvents)
    }

    @Test
    fun `test - multiple events - multiple collectors`() = runTest(Events()) {
        /* Collector A */
        val receivedEventsA = mutableListOf<TestCoroutineEventA>()
        collectEventsAsync<TestCoroutineEventA>(Dispatchers.Unconfined) { receivedEventsA += it }

        /* Collector B */
        val receivedEventsB = mutableListOf<TestCoroutineEventA>()
        collectEventsAsync<TestCoroutineEventA>(Dispatchers.Unconfined) { receivedEventsB += it }

        /* Collector C */
        val receivedEventsC = mutableListOf<TestCoroutineEventB>()
        collectEventsAsync<TestCoroutineEventB>(Dispatchers.Unconfined) { receivedEventsC += it }

        /* Send events */
        TestCoroutineEventA(1).emit()
        TestCoroutineEventB(2).emit()
        TestCoroutineEventA(3).emit()
        TestCoroutineEventB(4).emit()

        coroutineContext.cancelChildren()

        assertEquals(listOf(TestCoroutineEventA(1), TestCoroutineEventA(3)), receivedEventsA)
        assertEquals(listOf(TestCoroutineEventA(1), TestCoroutineEventA(3)), receivedEventsB)
        assertEquals(listOf(TestCoroutineEventB(2), TestCoroutineEventB(4)), receivedEventsC)
    }

    @OptIn(ExperimentalCoroutinesApi::class)
    @Test
    fun `test - slow subscriber does not block other emissions`() = runTest(Events()) {
        val receivedEventsA = mutableListOf<TestCoroutineEventA>()
        val receivedEventsB = mutableListOf<TestCoroutineEventB>()

        // Slow subscriber
        collectEventsAsync<TestCoroutineEventB>(context = UnconfinedTestDispatcher(testScheduler)) {
            delay(1000)
            receivedEventsB += it
        }


        collectEventsAsync<TestCoroutineEventA>(context = UnconfinedTestDispatcher(testScheduler)) {
            receivedEventsA += it
        }

        testScheduler.advanceUntilIdle()
        TestCoroutineEventA(1).emit()
        testScheduler.advanceUntilIdle()
        assertEquals(listOf(TestCoroutineEventA(1)), receivedEventsA)
        assertEquals(0, testScheduler.currentTime)

        TestCoroutineEventA(2).emit()
        assertEquals(listOf(TestCoroutineEventA(1), TestCoroutineEventA(2)), receivedEventsA)
        assertEquals(0, testScheduler.currentTime)

        TestCoroutineEventB(3).emit()
        assertEquals(listOf(TestCoroutineEventB(3)), receivedEventsB)
        assertEquals(1000, testScheduler.currentTime)

        coroutineContext.job.cancelChildren()
    }

    @Test
    @OptIn(ExperimentalCoroutinesApi::class)
    fun `test - slow and fast subscriber loops`() = runTest(Events()) {
        val receivedEventsA = mutableListOf<TestCoroutineEventA>()
        val receivedEventsB = mutableListOf<TestCoroutineEventB>()

        // fast
        collectEventsAsync<TestCoroutineEventA>(context = UnconfinedTestDispatcher(testScheduler)) {
            delay(1)
            receivedEventsA += it
        }

        // slow
        collectEventsAsync<TestCoroutineEventB>(context = UnconfinedTestDispatcher(testScheduler)) {
            delay(10)
            receivedEventsB += it
        }

        // Fast emissions
        val fast = launch(StandardTestDispatcher(testScheduler)) {
            repeat(1_000) {
                TestCoroutineEventA(it).emit()
            }
        }

        // Slow emissions
        val slow = launch(StandardTestDispatcher(testScheduler)) {
            repeat(1_000) {
                TestCoroutineEventB(it).emit()
            }
        }

        fast.join()
        assertEquals(buildList { repeat(1_000) { add(TestCoroutineEventA(it)) } }, receivedEventsA)
        assertEquals(TestCoroutineEventB(99), receivedEventsB.last())

        slow.join()
        assertEquals(TestCoroutineEventB(999), receivedEventsB.last())
        coroutineContext.job.cancelChildren()
    }

    @Test
    @OptIn(ExperimentalCoroutinesApi::class)
    fun `test - event emission waits for all subscribers to finish`() = runTest(Events()) {
        val receivedSubscriber1 = mutableListOf<TestCoroutineEventA>()
        val receivedSubscriber2 = mutableListOf<TestCoroutineEventA>()

        collectEventsAsync<TestCoroutineEventA>(context = UnconfinedTestDispatcher(testScheduler)) {
            receivedSubscriber1 += it
            delay(1)
        }

        collectEventsAsync<TestCoroutineEventA>(context = UnconfinedTestDispatcher(testScheduler)) {
            receivedSubscriber2 += it
            delay(10)
        }

        TestCoroutineEventA(1).emit()
        assertEquals(listOf(TestCoroutineEventA(1)), receivedSubscriber1)
        assertEquals(listOf(TestCoroutineEventA(1)), receivedSubscriber2)

        TestCoroutineEventA(2).emit()
        assertEquals(listOf(TestCoroutineEventA(1), TestCoroutineEventA(2)), receivedSubscriber1)
        assertEquals(listOf(TestCoroutineEventA(1), TestCoroutineEventA(2)), receivedSubscriber2)

        currentCoroutineContext().job.cancelChildren()
    }

    @OptIn(ExperimentalCoroutinesApi::class)
    @Test
    fun `test - event listeners are executed in parallel`() = runTest(Events()) {
        data class Checkpoint(val name: String)

        val collectorAWaiting = Checkpoint("CollectorAWaiting")
        val collectorBWaiting = Checkpoint("CollectorBWaiting")
        val collectorAFinished = Checkpoint("CollectorAFinished")
        val collectorBFinished = Checkpoint("CollectorBFinished")

        var checkpoint: Checkpoint? = null

        collectEventsAsync<TestCoroutineEventA>(context = UnconfinedTestDispatcher(testScheduler)) {
            assertEquals(null, checkpoint)
            checkpoint = collectorAWaiting
            delay(10)
            assertEquals(10, testScheduler.currentTime)
            checkpoint = collectorAFinished
        }

        collectEventsAsync<TestCoroutineEventA>(context = UnconfinedTestDispatcher(testScheduler)) {
            assertEquals(collectorAWaiting, checkpoint)
            checkpoint = collectorBWaiting
            delay(20)
            assertEquals(20, testScheduler.currentTime)

            assertEquals(collectorAFinished, checkpoint)
            checkpoint = collectorBFinished
        }

        TestCoroutineEventA(0).emit()
        assertEquals(collectorBFinished, checkpoint)
        currentCoroutineContext().job.cancelChildren()
    }

    @Test
    fun `test - exception`() = runTest(Events()) {
        val exceptions = mutableListOf<Throwable>()
        val exception = IllegalStateException("Test Exception")
        val handler = CoroutineExceptionHandler { _, throwable ->
            exceptions += throwable
        }

        collectEventsAsync<TestCoroutineEventA>(context = NonCancellable + StandardTestDispatcher(testScheduler) + handler) { event ->
            if (event == TestCoroutineEventA(10)) {
                throw exception
            }
        }

        /* Advance to ensure that the collector is active */
        testScheduler.advanceUntilIdle()
        TestCoroutineEventA(1).emit()
        TestCoroutineEventA(10).emit()

        testScheduler.advanceUntilIdle()
        assertSame(exception.message, exceptions.single().message)
    }

    @Test
    fun `test - listener gets cleaned in Events`() = runTest(Events()) {
        val events = currentCoroutineContext().eventsOrThrow as EventsImpl
        val collector = collectEventsAsync<TestCoroutineEventA> {}
        assertEquals(setOf(TestCoroutineEventA::class), events.typedChannels.snapshot().keys)

        collector.cancelAndJoin()
        assertEquals(emptySet<Any>(), events.typedChannels.snapshot().keys)
    }

    @Test
    fun `test - listener gets cleaned in Events - multiple collectors`() = runTest(Events()) {
        val events = currentCoroutineContext().eventsOrThrow as EventsImpl
        val collector1 = collectEventsAsync<TestCoroutineEventA> {}
        val collector2 = collectEventsAsync<TestCoroutineEventA> {}
        assertEquals(setOf(TestCoroutineEventA::class), events.typedChannels.snapshot().keys)
        assertEquals(2, events.typedChannels.snapshot()[TestCoroutineEventA::class]?.snapshot()?.size)

        collector1.cancelAndJoin()
        assertEquals(1, events.typedChannels.snapshot()[TestCoroutineEventA::class]?.snapshot()?.size)

        collector2.cancelAndJoin()
        assertEquals(emptySet<Any>(), events.typedChannels.snapshot().keys)
    }
}